package cn.azzhu.day05

import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
/**
 * @author azzhu
 * @create 2020-09-19 21:46:59
 */
object ScalaMapWithState {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.enableCheckpointing(5000)

    env.setStateBackend(new FsStateBackend("file:///D:\\bigdata\\flink-learning\\backend"))

    val lines = env.socketTextStream("hadoop105", 9999)

    val keyed = lines.flatMap(_.split(",")).map(word => {
      if(word.equals("storm")) {
        println(1/0)
      }
      (word,1)
    }).keyBy(0)

    val summed = keyed.mapWithState((input: (String, Int), state: Option[Int]) => {
      state match {
        case Some(count) => {
          val key = input._1
          val value = input._2
          val sum = count + value
          ((key, sum), Some(sum))
        }
        case None => {
          (input, Some(input._2))
        }
      }
    })

    summed.print()

    env.execute("ScalaMapWithState")
  }
}
